package com.zyh.flink.day02.transform;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class SideOutPutStreamTest {
    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<Integer> dataStream = environment.fromElements(1, 2, 3, 4, 5, 6, 7);
        /*
        * dataStream分流成奇数流和偶数流
        * 创建2个子流的标签
        * */
        OutputTag<Integer> evenTag = new OutputTag<>("even", Types.INT);
        OutputTag<Integer> oddTag = new OutputTag<>("odd", Types.INT);
        SingleOutputStreamOperator<Integer> dataStream2 = dataStream.process(new ProcessFunction<Integer, Integer>() {
            @Override
            public void processElement(Integer value, Context context, Collector<Integer> collector) throws Exception {
                if (value % 2 == 0) {
                    //context工具打标签
                    context.output(evenTag, value);
                } else {
                    context.output(oddTag, value);
                }
            }
        });
        /*
        * 拆分子流
        * */
        DataStream<Integer> evenStream = dataStream2.getSideOutput(evenTag);
        DataStream<Integer> oddStream = dataStream2.getSideOutput(oddTag);

        evenStream.print("even");
        oddStream.print("odd");

        environment.execute("SideJob");
    }
}
